package cn.cihon.ubi.qorosauto.dao.hadoop;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import cn.cihon.ubi.qorosauto.bo.VehicleStandardData;
import cn.cihon.ubi.qorosauto.util.DateUtil;

public class VehicleStandardDataHdfsDao {

	// 文件系统连接到 hdfs的配置信息
	private static Configuration getConf() {
		Configuration conf = new Configuration();
		conf.set("mapred.job.tracker", "123.57.54.113:9001");
		conf.set("fs.default.name", "hdfs://123.57.54.113:9000");
		conf.set("dfs.client.block.write.replace-datanode-on-failure.policy",
				"NEVER");
		conf.set("dfs.client.block.write.replace-datanode-on-failure.enable",
				"true");
		return conf;
	}

	/**
	 * 利用行车数据获得文件名称
	 * 
	 * @param stdData
	 * @return
	 */
	public String getFileNameByStdData(VehicleStandardData stdData) {
		// 获取时间戳
		Date recordTimestamp = stdData.getRecordTimestamp();
		String dateStr = DateUtil.getDateStr(recordTimestamp);
		// 获得 VIN 码
		String vin = stdData.getVIN();
		// 用时间戳和VIN码获得文件名
		String fileName = dateStr + vin + ".csv";
		return fileName;
	}

	/**
	 * 利用行车数据获得文件夹名称
	 * 
	 * @param stdData
	 * @return
	 */
	public String getDirNameByStdData(VehicleStandardData stdData) {
		// 获取时间戳
		Date recordTimestamp = stdData.getRecordTimestamp();
		String dateStr = DateUtil.getDateStr(recordTimestamp);
		// 将日期字符串当做文件夹名称返回
		return dateStr;
	}

	public String getFileRelativePath(String dirName, String fileName) {
		return "/cihon/" + dirName + "/" + fileName;
	}

	/**
	 * 在hdfs上创建文件夹
	 * 
	 * @param stdData
	 * @throws IOException
	 */
	public void mkdir(String path) throws IOException {
		Configuration conf = getConf();  
		FileSystem fs = FileSystem.get(conf);
		fs.mkdirs(new Path(path));
		fs.close();
	}

	/**
	 * 在Hdfs上创建新文件
	 * 
	 * @param fileName
	 * @throws IOException
	 */
	public void createNewHDFSFile(String fileName) throws IOException {
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		Path filePath = new Path(fileName);
		FSDataOutputStream out = fs.create(filePath);
		out.close();
		fs.close();
	}

	/**
	 * 判断文件夹或文件是否存在
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 */
	public Boolean isExist(String pathStr) throws IOException {
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathStr);
		Boolean isExist = fs.exists(path);
		return isExist;
	}

	/**
	 * 向文件追加文本数据
	 * 
	 * @param fileRelativePath
	 * @param strDataStr
	 * @throws IOException
	 */
	private void writeDataToHdfsFile(String fileRelativePath, String strDataStr)
			throws IOException {
		Configuration conf = getConf();
		FileSystem hdfs = FileSystem.get(conf);
		Path path = new Path(fileRelativePath);
		FSDataOutputStream out = hdfs.append(path);
		out.write(strDataStr.getBytes());
		out.close();
		hdfs.close();
	}

	/**
	 * 
	 * 向文件中追加明文行车数据
	 * 
	 * @throws FileNotFoundException
	 * @throws IOException
	 */
	public void writeStdDataToHdfs(VehicleStandardData stdData)
			throws IOException {

//		// 根据行车数据获得文件夹名称
//		String dirName = this.getDirNameByStdData(stdData);
//		// 判断文件夹是否存在
//		Boolean isDirExist = this.isExist(dirName);
//		if (!isDirExist) {
//			// 如果文件夹不存在，创建文件夹
//			this.mkdir(dirName);
//		}
//		// 根据行车数据获得文件名称
//		String fileName = this.getFileNameByStdData(stdData);
//		// 拼接文件夹和文件，获得文件在HDFS中的相对路径
//		String fileRelativePath = this.getFileRelativePath(dirName, fileName);
//		// 判断文件是否存在
//		Boolean isFileExist = this.isExist(fileRelativePath);
//		if (!isFileExist) {
//			// 如果文件不存在，创建文件
//			this.createNewHDFSFile(fileRelativePath);
//		}
		
		// 获得行车明文数据
		String strDataStr = stdData.toString() + "\r\n";
		// 向文件中追加行车明文数据
		writeDataToHdfsFile("/data.csv", strDataStr);

	}

	/**
	 * @Title: viewPath
	 * @Description: 列出目录下所有文件夹和文件
	 * @return void 返回类型
	 * @throws
	 */
	public void downloadHdfsFile(Path path) throws IOException {
		Configuration conf = getConf();
		FileSystem hdfs = FileSystem.get(conf);
		FileStatus stat = hdfs.getFileStatus(path);

		if (stat.isDirectory()) {
			System.out.println(stat.getPath());
			File file = new File(stat.getPath().toString()
					.replaceFirst("hdfs://123.57.54.113:9000/", "/home/yan/"));
			file.mkdir();
			FileStatus[] stats = hdfs.listStatus(path);
			for (int i = 0; i < stats.length; i++) {
				downloadHdfsFile(stats[i].getPath());
			}
		} else {
			System.out.println("文件名为：" + stat.getPath().getName());
			FSDataInputStream fsdi = hdfs.open(stat.getPath());
			OutputStream output = new FileOutputStream(stat.getPath().toString()
					.replaceFirst("hdfs://123.57.54.113:9000/", "/home/yan/"));
			IOUtils.copyBytes(fsdi, output, 4096, true);
		}
		hdfs.close();
	}
	
	/*
	 * delete a dir in the hdfs
	 * dir may like '/tmp/testdir'
	 */
	public void deleteDir(String dir) throws IOException
	{
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		fs.delete(new Path(dir));
		fs.close();
	}

	public static void main(String[] args) throws FileNotFoundException,
			IOException {
		VehicleStandardDataHdfsDao dao = new VehicleStandardDataHdfsDao();
		System.out.println("开始读取");
//		Configuration conf = getConf();
//		FileSystem fs = FileSystem.get(conf);
//		Path filePath = new Path("/data.csv");
//		FSDataOutputStream out = fs.create(filePath);
//		out.close();
//		fs.close();
//		dao.deleteDir("/cihon");
//		dao.deleteDir("/user");
		dao.downloadHdfsFile(new Path("/"));
		System.out.println("读取结束");
	}

}